package com.chukun.flink.stream.action.fraud.rules.source;


import com.chukun.flink.stream.action.fraud.config.Config;
import com.chukun.flink.stream.action.fraud.config.Parameters;
import java.util.Properties;

import com.chukun.flink.stream.action.fraud.rules.functions.JsonDeserializer;
import com.chukun.flink.stream.action.fraud.rules.functions.JsonGeneratorWrapper;
import com.chukun.flink.stream.action.fraud.rules.functions.TimeStamper;
import com.chukun.flink.stream.action.fraud.rules.functions.TransactionsGenerator;
import com.chukun.flink.stream.action.fraud.rules.model.Transaction;
import com.chukun.flink.stream.action.fraud.rules.utils.KafkaUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

/**
 * 交易数据流
 * @author chukun
 */
public class TransactionsSource {

  /**
   * 创建交易数据流
   * @param config
   * @return
   */
  public static SourceFunction<String> createTransactionsSource(Config config) {

    String sourceType = config.get(Parameters.TRANSACTIONS_SOURCE);
    Type transactionsSourceType =
        Type.valueOf(sourceType.toUpperCase());

    if (transactionsSourceType == Type.KAFKA) {
      Properties kafkaProps = KafkaUtils.initConsumerProperties(config);
      String transactionsTopic = config.get(Parameters.DATA_TOPIC);
      FlinkKafkaConsumer011<String> kafkaConsumer =
          new FlinkKafkaConsumer011<>(transactionsTopic, new SimpleStringSchema(), kafkaProps);
      kafkaConsumer.setStartFromLatest();
      return kafkaConsumer;
    } else {
      int transactionsPerSecond = config.get(Parameters.RECORDS_PER_SECOND);
      return new JsonGeneratorWrapper<>(new TransactionsGenerator(transactionsPerSecond));
    }
  }

  public static DataStream<Transaction> stringsStreamToTransactions(
      DataStream<String> transactionStrings) {
    return transactionStrings
        .flatMap(new JsonDeserializer<>(Transaction.class))
        .returns(Transaction.class)
        .flatMap(new TimeStamper<>())
        .returns(Transaction.class)
        .name("Transactions Deserialization");
  }

  public enum Type {
    GENERATOR("Transactions Source (generated locally)"),
    KAFKA("Transactions Source (Kafka)");

    private String name;

    Type(String name) {
      this.name = name;
    }

    public String getName() {
      return name;
    }
  }
}
